package alfred.mdp.broker;

import alfred.mdp.messages.client.Request;
import alfred.mdp.utils.Shutdowner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;
import org.zeromq.ZMQException;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentMap;

import static alfred.mdp.utils.Utils.asString;

/**
 */
public final class WorkerReferenceDispatcher implements Runnable {
    private static final Logger LOG = LoggerFactory.getLogger(WorkerReferenceDispatcher.class);
    private static final String DEFAULT_LOG_MESSAGE = "WorkerReferenceDispatcher of worker {} and service {} - ";
    private WorkerReference workerReference;
    private Broker broker;

    public WorkerReferenceDispatcher(Broker broker, WorkerReference workerReference) {
        this.broker = broker;
        this.workerReference = workerReference;
    }

    @Override
    public void run() {
        String serviceName = workerReference.getServiceName();
        while (!Thread.currentThread().isInterrupted()) {
            try {
                if (LOG.isTraceEnabled()) {
                    LOG.trace(DEFAULT_LOG_MESSAGE + "find request", asString(workerReference.getWorkerId()), serviceName);
                }
                Request request;
                ConcurrentMap<String, BlockingQueue<Request>> requestsForService = broker.getRequestsForService();
                BlockingQueue<Request> requests = requestsForService.get(serviceName);
                while ((request = requests.take()) != null) {
                    if (LOG.isDebugEnabled()) {
                        LOG.debug(DEFAULT_LOG_MESSAGE + "found request", asString(workerReference.getWorkerId()), serviceName);
                    }
                    broker.sendMessageToWorker(new alfred.mdp.messages.worker.Request(request.getClientId(), request.getBody()), workerReference.getWorkerId().array());
                    }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();

            } catch (ZMQException e) {
                LOG.error("ZMQException", e);
                // context destroyed, exit
                if (ZMQ.Error.ETERM.getCode() == e.getErrorCode()) {
                    break;
                }
                throw e;

            } catch (Exception e) {
                LOG.error("Error", e);

            }
        }
        Shutdowner.shutdownCompleted(workerReference.getShutdownGroup(), getClass());
        LOG.debug(DEFAULT_LOG_MESSAGE + "shutdown completed", asString(workerReference.getWorkerId()), serviceName);
    }
}
